
package com.ys.chatserver.im.impl;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson2.JSON;
import com.eva.epc.common.util.CommonUtils;
import com.eva.framework.dbpool.DBDepend;
import com.eva.framework.utils.LoggerFactory;
import com.google.gson.Gson;
import com.ys.chatserver.BaseConf;
import com.ys.chatserver.MyControllerJSON;
import com.ys.chatserver.common.dto.cnst.OperateLog;
import com.ys.chatserver.common.dto.cnst.RtcRedisKeyConstants;
import com.ys.chatserver.common.dto.cnst.UserProtocalsType;
import com.ys.chatserver.http.logic.LogicProcessor2;
import com.ys.chatserver.im.ChatServerEventListener;
import com.ys.chatserver.im.ChatServerLauncher;
import com.ys.chatserver.im.dto.*;
import com.ys.chatserver.im.util.ChatServerUtils;
import com.ys.chatserver.im.util.MessageHelper;
import com.ys.chatserver.im.util.MessageHelper.SendResultObserver;
import com.ys.chatserver.redis.RedisConfig;
import com.ys.chatserver.redis.RedisManager;
import com.ys.chatserver.redis.RedisUtil;
import com.ys.chatserver.rtc.ConstRtcApp;
import com.ys.chatserver.tool.EmptyUtils;
import io.netty.channel.Channel;
import net.ys.mobileimsdk.server.protocal.Protocal;
import org.slf4j.Logger;
import redis.clients.jedis.Jedis;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * IM Server 服务器通用业务逻辑实现类（除群聊外的各种逻辑）.
 *
 * @author Jack Jiang(http://www.52im.net/space-uid-1.html)
 * @version 1.0
 */
public class ChatLogicManager {

    static final Logger logger = LoggerFactory.getLog();

    static final AtomicInteger num = new AtomicInteger();

    static final Set<String> group_channelIds = Collections.synchronizedSet(new HashSet<>());

    static final Set<String> sosChannelIds = Collections.synchronizedSet(new HashSet<>());

    static final Object lock = new Object();

    static final Semaphore semphore = new Semaphore(30);

    static final ConcurrentHashMap<String, CallResDto> callOutimeMap = new ConcurrentHashMap<>();

    static final ConcurrentHashMap<String, Long> callTimeMap = new ConcurrentHashMap<>();

    public ChatLogicManager() {
        //
    }

    // 处理临时聊天消息的最终投递
    public static boolean processMT42_OF_TEMP$CHAT$MSG_A$TO$SERVER(String fingerPrint, String dataContent) throws Exception {
        UsersStatusManager liverUsersMgr = ChatServerEventListener.getInstance().getLiverUsersMgr();

        // 解析临时聊天消息
        final MsgBody4Guest tempChatMsgFromClient = MessageHelper.pareseTempChatMsg_A$TO$SERVER_Message(dataContent);
        tempChatMsgFromClient.setM3(String.valueOf(System.currentTimeMillis()));
        // 结果正常解析出来了
        // 目标用户是否在线（准确地说是否在线于APP的IM服务端）
        String destUser_id = tempChatMsgFromClient.getT();
        final boolean isDestFriendOnline = liverUsersMgr.isOnline(destUser_id);

        // 指令发送结果观察者（因为通信底层netty是异步通知数据发送结果的）
        SendResultObserver sendResultObserver = (code, p) -> {
            // 返回值true：对于应用层逻辑来说，无论是在线发送成功，或者是跨服桥接成功，都算是发送成功
            if (code) {
                LoggerFactory.getLog().debug("[RBChat]服务器通知由uid=" + tempChatMsgFromClient.getF()
                        + "发起的临时聊天消息，已经成功发给在线好友uid=" + tempChatMsgFromClient.getT() + ".");

                // 说明：满足此条件，表示本地不在线但处于与Web互通模式的情况下，消息却发送成功，也需要进行App的push通知，
                //       因为前提是app端已不在线，进行push是合情合理的（否则桥接成功的情况下，app这边即使不在线也永远收
                //       不到app的push通知，除非web那端也要实现不在线时的app的push通知。
                // 备忘：此处的push逻辑是为了省去web那端不在线时的app push通知逻辑，日后支持全功能集群后，就不需
                //       要这个逻辑了，因为包括web集群实例在内，都将是对等的且必须支持app 的push逻辑——所以没必要以下代码。
                if (!isDestFriendOnline && ChatServerLauncher.bridgeEnabled) {
                    // 尝试进行app端的push通知
                    OfflineMessageManager.processOfflineMessage(p, true, tempChatMsgFromClient.getM3()); // 注意此参数为true！
                }
            }
            // 返回值false：实时或桥接发送没有成功
            else {
                LoggerFactory.getLog().warn("[RBChat]服务器实时或桥接通知由uid=" + tempChatMsgFromClient.getF()
                        + "发起的临时聊天消息没有成功（目标uid=" + tempChatMsgFromClient.getT() + "），马上考虑做离线处理！");

                // 将该条临时聊天消息插入到数据库中（以备下次接收方B上线后拿到）
                OfflineMessageManager.processOfflineMessage(p, false, tempChatMsgFromClient.getM3());
                ;
            }
        };

        // 将临时聊天消息实时转发给本次消息接收者（如果用户连在本APPP实例则直发，否则开启与web互通时将桥接发）
        MessageHelper.sendTempChatMsg_SERVER$TO$B_Message(destUser_id, tempChatMsgFromClient, fingerPrint, sendResultObserver);

        /** 20190328 by Jack Jiang：按照目前的设计原则，为了减化架构和处理的复杂性，服务端主动发起
         的S2C类型消息或指令一律由发送者处理（详见设计备忘），切记！ */
        // ** 【用户消息收集】收集用户的聊天消息，以供后台进行用户行为分析，
        MessageRecordManager.colectChattingMsgAsync(dataContent, fingerPrint);

        return true;
    }

    public static void processUserLoginAction(final String userId, String extra, Channel session) {
        UsersStatusManager liverUsersMgr = ChatServerEventListener.getInstance().getLiverUsersMgr();

        // 将该用户加入到在线用户列表
        liverUsersMgr.userOnline(userId);
        // TODO 用于debug时，生产时可以去掉哦！
        liverUsersMgr.printLiveUsers();
        // ** 用户登陆成功后由服务端通知他的好友，他上线了，此时就把它的user_id传到服务端了！
        // 在线用户列表
        ConcurrentMap<String, Channel> liveUsers =
                ChatServerEventListener.getInstance().getLiverUsersMgr().getLiveUsers();
        try {
            int userType = ChatServerUtils.getUserType(userId);
            // 该用户的好友列表
            Vector<Vector<?>> roster = ChatServerUtils.getRosterByUid(userId);

            // 遍历好友列表，看看它有哪些好友是在线的
            for (Vector<?> friendRow : roster) {
                // 好友uid
                final String friend_user_uid = (String) friendRow.get(0);
                final String friend_user_mail = (String) friendRow.get(1);

//				// 自已的uid（顺便取出来备用）
//				String user_uid = (String)friendRow.get(2);

                // FIXME: 备忘-> 因无法目前无全功能集群支持，无法取得所有im接入层的在线列表，所以
                //             当开通与web端互通时，就无法将上、下线通知给跟机器端了。不过，在线状态
                //             这种很难维护的东西，可以考虑像微信一样，在以后的版本中逐渐去掉，实际上来说
                //             ，用处也不是很大，只是传统IM（比如QQ）时代遗留下来的习惯而已！

                // 遍历在线好友列表，看看该好友是否在线
                if (liveUsers.containsKey(friend_user_uid)) {
                    // 指令发送结果观察者（因为通信底层netty是异步通知数据发送结果的）
                    SendResultObserver sendResultObserver = new SendResultObserver() {
                        @Override
                        public void update(boolean code, Protocal p) {
                            if (userType == 1) {
                                if (code)
                                    LoggerFactory.getLog().debug("[RBChat]》登陆设备" + userId + "向用户"
                                            + friend_user_uid + "/" + friend_user_mail + "发送上线通知成功！");
                                else
                                    LoggerFactory.getLog().debug("[RBChat]》登陆用户" + userId + "向用户"
                                            + friend_user_uid + "/" + friend_user_mail + "发送上线通知失败，错误code=" + code);

                            } else {
                                if (code)
                                    LoggerFactory.getLog().debug("[RBChat]》登陆用户" + userId + "向它的好友"
                                            + friend_user_uid + "/" + friend_user_mail + "发送上线通知成功！");
                                else
                                    LoggerFactory.getLog().debug("[RBChat]》登陆用户" + userId + "向它的好友"
                                            + friend_user_uid + "/" + friend_user_mail + "发送上线通知失败，错误code=" + code);
                            }
                        }
                    };
                    if (userType == 1) {
                        MessageHelper.sendDevOnlineMessage(friend_user_uid, JSONObject.toJSONString(userId), sendResultObserver);
                    } else {
                        MessageHelper.sendOnlineMessage(friend_user_uid, JSONObject.toJSONString(userId), sendResultObserver);
                    }
                } else {
                    if (userType == 1) {
                        LoggerFactory.getLog().debug("[RBChat]》登陆设备" + userId + "的可查看用户"
                                + friend_user_uid + "/" + friend_user_mail + "不在线，无需发送上线通知。");

                    } else {
                        LoggerFactory.getLog().debug("[RBChat]》登陆用户" + userId + "的好友"
                                + friend_user_uid + "/" + friend_user_mail + "不在线，无需发送上线通知。");
                    }
                }
            }
        } catch (Exception e) {
            LoggerFactory.getLog().warn(e.getMessage(), e);
        }
    }

    public static void processUserLogoutAction(String userId, Channel session, int beKickoutCode) {
        // 在线用户列表
        UsersStatusManager liverUsersMgr = ChatServerEventListener.getInstance().getLiverUsersMgr();

        // ** 用户注销后由服务端通知他的好友，他下线了！
        // 下线的用户，必然之前已经在在线列表里了，因为通过它的user_id就能查到它登陆时的uid或者mail了
        if (userId != null) {
            try {
                int userType = ChatServerUtils.getUserType(userId);
                // 该用户的好友列表
                Vector<Vector<?>> roster = ChatServerUtils.getRosterByUid(userId);

                // 遍历好友列表，看看它有哪些好友是在线的
                for (Vector<?> friendRow : roster) {
                    // 好友uid
                    final String friend_user_uid = (String) friendRow.get(0);
                    final String friend_user_mail = (String) friendRow.get(1);
                    // 自已的uid（顺便取出来备用）
                    //final String user_uid = (String)friendRow.get(2);

                    // FIXME: 备忘-> 因无法目前无全功能集群支持，无法取得所有im接入层的在线列表，所以
                    //             当开通与web端互通时，就无法将上、下线通知给跟机器端了。不过，在线状态
                    //             这种很难维护的东西，可以考虑像微信一样，在以后的版本中逐渐去掉，实际上来说
                    //             ，用处也不是很大，只是传统IM（比如QQ）时代遗留下来的习惯而已！

                    // 看看该好友是否在线
//					if(liveUsers.containsKey(friend_user_uid))
                    if (liverUsersMgr.isOnline(friend_user_uid)) {
                        // 指令发送结果观察者（因为通信底层netty是异步通知数据发送结果的）
                        SendResultObserver sendResultObserver = new SendResultObserver() {
                            @Override
                            public void update(boolean code, Protocal p) {

                                if (userType == 1) {
                                    if (code)
                                        LoggerFactory.getLog().debug("[RBChat]》登陆设备" + userId + "向用户"
                                                + friend_user_uid + "/" + friend_user_mail + "发送下线通知成功！");
                                    else
                                        LoggerFactory.getLog().debug("[RBChat]》登陆用户" + userId + "向用户"
                                                + friend_user_uid + "/" + friend_user_mail + "发送下线通知失败，错误code=" + code);

                                } else {
                                    if (code)
                                        LoggerFactory.getLog().debug("[RBChat]》登陆用户" + userId + "向它的好友"
                                                + friend_user_uid + "/" + friend_user_mail + "发送下线通知成功！");
                                    else
                                        LoggerFactory.getLog().debug("[RBChat]》登陆用户" + userId + "向它的好友"
                                                + friend_user_uid + "/" + friend_user_mail + "发送下线通知失败，错误code=" + code);
                                }
                            }
                        };
                        if (userType == 1) {
                            MessageHelper.sendDevOfflineMessage(friend_user_uid, JSONObject.toJSONString(userId), sendResultObserver);
                        } else {
                            MessageHelper.sendOfflineMessage(friend_user_uid, JSONObject.toJSONString(userId), sendResultObserver);
                        }
                    } else {
                        if (userType == 1) {
                            LoggerFactory.getLog().debug("[RBChat]》登陆设备" + userId + "的可查看用户"
                                    + friend_user_uid + "/" + friend_user_mail + "不在线，无需发送下线通知。");

                        } else {
                            LoggerFactory.getLog().debug("[RBChat]》登陆用户" + userId + "的好友"
                                    + friend_user_uid + "/" + friend_user_mail + "不在线，无需发送下线通知。");
                        }
                    }
                }
            } catch (Exception e) {
                LoggerFactory.getLog().warn(e.getMessage(), e);
            }
        } else {
            LoggerFactory.getLog().error("[RBChat]当前正在离线的用户" + userId + "为null？？？");
        }

        // 将该用户从在线用户列表中去掉
        liverUsersMgr.userOffline(userId);

        //增加退出系统日志
        OperateLog oprLog = new OperateLog(OperateLog.OPERATE_LOG_TYPE_LOGOUT_IM_SERVER
                , userId
                , "");
        try {
            MyControllerJSON.addOperateLog(oprLog);
        } catch (Exception e) {
            LoggerFactory.getLog().warn(e.getMessage(), e);
        }

        // TODO 用于debug时，生产时可以去掉哦！
        liverUsersMgr.printLiveUsers();
    }

    public static void processMT55_OF_SOS_NOTIFICATION$REQ_B$TO$Message(CMDSosRequest sosRequest) throws Exception {
        System.out.println("number is " + (num.incrementAndGet()));
        try {
            semphore.acquire();

            synchronized (lock) {
                if (group_channelIds.contains(sosRequest.getT())) {
                    sendGrabFail(sosRequest.getF());
                    logger.error("1-------------------------{} release lock {}----------------------------", sosRequest.getF(), sosRequest.getT());
                    return;
                } else {
                    group_channelIds.add(sosRequest.getT());
                }
            }
            logger.error("------------------------{} release acquire {}----------------------------", sosRequest.getF(), sosRequest.getT());
            synchronized (lock) {
                group_channelIds.remove(sosRequest.getT());
                logger.error("-------------------------{} get lock {}----------------------------", sosRequest.getF(), sosRequest.getT());
                boolean sos = RedisUtil.exists("sos-" + sosRequest.getT());
                boolean onekey = RedisUtil.exists("onekey-" + sosRequest.getT());
                if (sos || onekey) {
                    sendGrabFail(sosRequest.getF());
                    logger.error("2-------------------------{} release lock {}----------------------------", sosRequest.getF(), sosRequest.getT());
                    return;
                }
                String[] split = sosRequest.getT().split("_");
                String groupId = split[0];
                String channelId = split[1];
                String pttKey = RtcRedisKeyConstants.GET_PTT_GROUP_KEY(groupId, channelId);
                String sosKey = RtcRedisKeyConstants.GET_SOS_GROUP_KEY(groupId, channelId);

                if (RedisUtil.exists(sosKey)) {
                    sendGrabFail(sosRequest.getF());
                    logger.error("3-------------------------{} release lock {}----------------------------", sosRequest.getF(), sosRequest.getT());
                    return;
                }
                if (RedisUtil.exists(pttKey) && !RedisUtil.get(pttKey).startsWith(sosRequest.getF())) {
                    sendGrabFail(sosRequest.getF());
                    logger.error("3-------------------------{} release lock {}----------------------------", sosRequest.getF(), sosRequest.getT());
                    return;
                }
                Set<String> keys = RedisUtil.keys(RtcRedisKeyConstants.KEYS_ONE_GROUP(groupId));
                if (!keys.isEmpty()) {
                    List<String> levels = keys.stream().map(a -> a.split(":")[3]).collect(Collectors.toList());
                    Vector<Vector<?>> levelVector = LogicProcessor2.db.queryData("select level from chat_group_node where g_status = 1 and node_pid = '" + channelId + "' and node_type = 'USER' and user_uid = '" + sosRequest.getF() + "'");
                    String userLevel = levelVector.get(0).get(0).toString();
                    long count = levels.stream().filter(a -> a.compareTo(userLevel) < 0).count();
                    if (count > 0) {
                        sendGrabFail(sosRequest.getF());
                        logger.error("4-------------------------{} release lock {}----------------------------", sosRequest.getF(), sosRequest.getT());
                        return;
                    }
                }
                RedisUtil.set(pttKey, sosRequest.getF(), Long.parseLong(BaseConf.getInstance().getREDIS_PTT_TTL()));
                final long messageTime = System.currentTimeMillis();
                final String urlData = ConstRtcApp.USER + "-" + groupId + "_" + channelId + "-" + channelId + "-" + sosRequest.getF() + "-" + messageTime;
                logger.error("-------------------------{} send succ {}----------------------------", sosRequest.getF(), sosRequest.getT());
                sendPtt(groupId, sosRequest.getM(), sosRequest.getF(), getUserName(sosRequest.getF()), pttKey, urlData);
                new Thread(() -> {
                    String a = "ptt-" + sosRequest.getT() + "-" + sosRequest.getF() + "-" + messageTime;
                    if (!RedisUtil.exists(a)) {
                        RedisUtil.set(a, String.valueOf(messageTime), 5);
                    }
                    notifyCloseOutTimeChannel(a, urlData);
                }).start();
                logger.error("5-------------------------{} release lock {}----------------------------", sosRequest.getF(), sosRequest.getT());
            }
        } catch (InterruptedException e) {
            logger.error("------------------------{} interrupted {}----------------------------", sosRequest.getF(), sosRequest.getT());
            Thread.currentThread().interrupt();
        } finally {
            logger.error("------------------------{} release semphore {}----------------------------", sosRequest.getF(), sosRequest.getT());
            semphore.release();
        }
    }

    private static void sendGrabFail(String usrId) throws Exception {
        System.out.println("<<<<<<<<<" + usrId);
        PushResDTO dto = new PushResDTO();
        dto.setType(NotifyEnum.THREE.getType());
        String json = new Gson().toJson(dto);
        MessageHelper.processMT55_OF_SOS_NOTIFICATION$REQ_B$TO$Message(usrId, json, null);
    }

    private static void sendPtt(String groupId, String channelId, String userId, String name, String key, String urlData) throws Exception {
        System.out.println(">>>>>>>:" + userId);
        String sql = "select user_uid from chat_group_node where g_status = 1 and node_pid = '" + channelId + "' and base_id = '" + groupId + "' and node_type = 'USER'";
        Vector<Vector<?>> vectors = LogicProcessor2.db.queryData(sql);
        Map<String, AtomicInteger> numberMap = UserProtocalsType.SOS_NUMBER_MAP;
        if (!vectors.isEmpty()) {
            PushResDTO dto = new PushResDTO();
            dto.setType(NotifyEnum.TWO.getType());
            dto.setUrlData(urlData);
            dto.setAppType(ConstRtcApp.APP_PTT);
            dto.setName(name);
            dto.setPushId(userId);
            dto.setGroupId(groupId + "_" + channelId);
            dto.setChannelId(channelId);
            String json = new Gson().toJson(dto);
            List<String> exist = new ArrayList<>();
            for (Vector<?> vector : vectors) {
                String str = vector.get(0).toString();
                if (userId.equals(str)) {
                    continue;
                }
                exist.add(str);
                MessageHelper.processMT55_OF_SOS_NOTIFICATION$REQ_B$TO$Message(str, json, null);
            }
            String sids = "select s_ids from chat_group_node where node_id = '" + channelId + "' and base_id = '" + groupId + "' and g_status = 1";
            Vector<Vector<?>> sidres = LogicProcessor2.db.queryData(sids);
            Object susers = sidres.get(0).get(0);
            int size = vectors.size();
            if (EmptyUtils.isNotEmpty(susers)) {
                String substring = susers.toString().substring(1, susers.toString().lastIndexOf(","));
                List<String> suserIds = Stream.of(substring.split(",")).collect(Collectors.toList());
                List<String> suser = suserIds.stream().filter(f -> !exist.contains(f)).collect(Collectors.toList());
                size = suser.size() + vectors.size();
                for (String str : suser) {
                    if (userId.equals(str)) {
                        continue;
                    }
                    MessageHelper.processMT55_OF_SOS_NOTIFICATION$REQ_B$TO$Message(str, json, null);
                }
            }
            //waitNotify(key, size, numberMap);
        }

        PushResDTO dto = new PushResDTO();
        dto.setType(NotifyEnum.ONE.getType());
        dto.setUrlData(urlData);
        dto.setAppType(ConstRtcApp.APP_PTT);
        dto.setSuccessMsg("抢占成功!");
        dto.setChannelId(channelId);
        dto.setPushId(userId);
        String json = new Gson().toJson(dto);
        MessageHelper.processMT55_OF_SOS_NOTIFICATION$REQ_B$TO$Message(userId, json, null);
        numberMap.remove(key);
    }

    private static void notifyCloseOutTimeChannel(String key, String urlData) {
        System.out.println("key:" + key);
        while (true) {
            if (!RedisUtil.exists(key)) {
                break;
            }
            String s = RedisUtil.get(key);
            if (s != null && System.currentTimeMillis() - Long.parseLong(s) >= 5 * 1000) {
                try {
                    RedisUtil.del(key);
                    String[] split = key.split("-");
                    CMDSosRequest cmdSosRequest = new CMDSosRequest();
                    cmdSosRequest.setT(split[1]);
                    cmdSosRequest.setF(split[2]);
                    cmdSosRequest.setM(split[1].split("_")[1]);
                    ChatLogicManager.processMT56_OF_SOS_NOTIFICATION_SUCCESS_REMOVE_KEY(cmdSosRequest, urlData, true);
                    break;
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                }
            }
        }
    }

    private static void notifyCloseOutTimeSosChannel(String key, String urlData) {
        logger.debug("-------------urlData:{}----------", key);
        while (true) {
            if (!RedisUtil.exists(key)) {
                break;
            }
            String s = RedisUtil.get(key);
            if (s != null && System.currentTimeMillis() - Long.parseLong(s) >= 5 * 1000) {
                try {
                    RedisUtil.del(key);
                    String[] split = key.split("-");
                    CMDSosRequest cmdSosRequest = new CMDSosRequest();
                    cmdSosRequest.setT(split[1]);
                    cmdSosRequest.setF(split[2]);
                    ChatLogicManager.processMT63_OF_SOS_ALL_NOTIFICATION_SUCCESS_REMOVE_KEY(cmdSosRequest, urlData);
                    break;
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                }
            }
        }
    }

    public static void processMT56_OF_SOS_NOTIFICATION_SUCCESS_REMOVE_KEY(CMDSosRequest sosRequest, String urlData, Boolean outime) throws Exception {
        String userId = sosRequest.getF();
        String groupId = sosRequest.getT();
        String[] arr = groupId.split("_");
        String sql = "select user_uid from chat_group_node where base_id ='" + arr[0] + "' and node_pid = '" + arr[1] + "' and node_type = 'USER'";
        Vector<Vector<?>> vectors = LogicProcessor2.db.queryData(sql);
        List<String> exist = new ArrayList<>();
        if (!Objects.isNull(vectors) && vectors.size() > 1) {
            for (Vector<?> vector : vectors) {
                String str = vector.get(0).toString();
                if (!outime && userId.equals(str)) {
                    continue;
                }
                exist.add(str);
                PushResDTO dto = new PushResDTO();
                // dto.setType(outime && userId.equals(str) ? NotifyEnum.EIGHTEEN.getType() : NotifyEnum.FOUR.getType());
                dto.setType(NotifyEnum.FOUR.getType());
                dto.setGroupId(groupId);
                dto.setChannelId(arr[1]);
                dto.setPushId(userId);
                dto.setUrlData(urlData);
                String json = new Gson().toJson(dto);
                MessageHelper.processMT56_OF_SOS_NOTIFICATION_SUCCESS$REQ_B$TO$Message(str, json, null);
            }
        }

        String sids = "select s_ids from chat_group_node where node_id = '" + arr[1] + "' and base_id = '" + arr[0] + "' and g_status = 1";
        Vector<Vector<?>> sidres = LogicProcessor2.db.queryData(sids);
        Object susers = sidres.get(0).get(0);
        if (EmptyUtils.isNotEmpty(susers)) {
            String substring = susers.toString().substring(1, susers.toString().lastIndexOf(","));
            List<String> suserIds = Stream.of(substring.split(",")).collect(Collectors.toList());
            List<String> suser = suserIds.stream().filter(f -> !exist.contains(f)).collect(Collectors.toList());
            for (String str : suser) {
                if (userId.equals(str)) {
                    continue;
                }
                PushResDTO dto = new PushResDTO();
                dto.setType(NotifyEnum.FOUR.getType());
                dto.setGroupId(groupId);
                dto.setChannelId(arr[1]);
                dto.setPushId(userId);
                dto.setUrlData(urlData);
                String json = new Gson().toJson(dto);
                MessageHelper.processMT56_OF_SOS_NOTIFICATION_SUCCESS$REQ_B$TO$Message(str, json, null);
            }
        }

        String value = RedisUtil.get(RtcRedisKeyConstants.GET_PTT_GROUP_KEY(arr[0], arr[1]));
        if (!Objects.isNull(value)) {
            String[] strings = value.split("-");
            if (userId.equals(strings[0])) {
                RedisUtil.del(RtcRedisKeyConstants.GET_PTT_GROUP_KEY(arr[0], arr[1]));
            }
        }
    }

    public static void processMT57_OF_SOS_ALL_NOTIFICATION$REQ_B$TO$Message(CMDSosRequest sosRequest) throws Exception {
        String[] split = sosRequest.getT().split("_");
        String groupId = split[0];
        String channelId = split[1];
        String userId = sosRequest.getF();

        if (sosChannelIds.contains(channelId)) {
            sendSosFail(userId, sosRequest);
            return;
        }
        boolean onekey = RedisUtil.exists("onekey-" + sosRequest.getT());
        boolean sos = RedisUtil.exists("sos-" + sosRequest.getT());
        if (onekey || sos) {
            // 抢占失败
            sendSosFail(userId, sosRequest);
            return;
        }
        Set<String> keys = RedisUtil.keys(RtcRedisKeyConstants.KEYS_ONE_GROUP(groupId));
        if (!keys.isEmpty()) {
            List<String> levels = keys.stream().map(a -> a.split(":")[3]).collect(Collectors.toList());
            Vector<Vector<?>> levelVector = LogicProcessor2.db.queryData("select level from chat_group_node where g_status = 1 and node_id = '" + channelId + "' and node_type = 'MAIN_CHANNEL' ");
            String userLevel = levelVector.get(0).get(0).toString();
            long count = levels.stream().filter(a -> a.compareTo(userLevel) <= 0).count();
            if (count > 0) {
                // 抢占失败
                sendSosFail(userId, sosRequest);
                return;
            }
        }
        if (sosChannelIds.contains(channelId)) {
            sendSosFail(userId, sosRequest);
            return;
        }
        String val = RedisUtil.get(RtcRedisKeyConstants.GET_SOS_GROUP_KEY(groupId, channelId));
        if (CommonUtils.isStringEmpty(val)) {
            if (sosChannelIds.contains(channelId)) {
                sendSosFail(userId, sosRequest);
                return;
            }
            if (sosChannelIds.contains(channelId)) {
                sendSosFail(userId, sosRequest);
                return;
            }
            if (sosChannelIds.contains(channelId)) {
                sendSosFail(userId, sosRequest);
                return;
            }
            if (!sosChannelIds.contains(channelId)) {
                sosChannelIds.add(channelId);
            }
            String sql = "select user_uid from chat_group_node where base_id = '" + groupId + "' and node_pid = '" + channelId + "' and g_status = 1 and node_type = 'USER'";
            Vector<Vector<?>> vectors = LogicProcessor2.db.queryData(sql);
            String name = getUserName(userId);
            long messageTime = System.currentTimeMillis();
            Set<String> keySet = RedisUtil.keys(RtcRedisKeyConstants.SOS_GROUP_KEYS);
            List<String> list = keySet.stream().filter(k -> k.contains(groupId) && k.contains(channelId)).collect(Collectors.toList());
            if (list.isEmpty()) {
                try {
                    RedisUtil.set("sos-" + sosRequest.getT(), "temp", 3L);
                    PushResDTO dto = new PushResDTO();
                    String urlData = ConstRtcApp.USER + "-SOS-" + groupId + "_" + channelId + "-" + channelId + "-" + userId + "-" + messageTime;
                    dto.setUrlData(urlData);
                    dto.setAppType(ConstRtcApp.APP_PTT);
                    dto.setName(name);
                    dto.setChannelId(channelId);
                    dto.setPushId(userId);
                    dto.setGroupId(groupId + "_" + channelId);
                    dto.setType(NotifyEnum.TWELVE.getType());
                    String json = new Gson().toJson(dto);

                    for (Vector<?> vector : vectors) {
                        String str = vector.get(0).toString();
                        if (userId.equals(str)) {
                            continue;
                        }
                        MessageHelper.processMT57_OF_SOS_ALL_NOTIFICATION$REQ_B$TO$Message(str, json, null);
                    }

                    Map<String, AtomicInteger> numberMap = UserProtocalsType.SOS_NUMBER_MAP;
                    waitNotify(groupId, vectors.size(), numberMap);

                    PushResDTO dto1 = new PushResDTO();
                    dto1.setUrlData(urlData);
                    dto1.setAppType(ConstRtcApp.APP_PTT);
                    dto1.setType(NotifyEnum.THIRTEEN.getType());
                    dto1.setChannelId(channelId);
                    dto1.setGroupId(sosRequest.getT());
                    String json1 = new Gson().toJson(dto1);
                    numberMap.remove(groupId);
                    MessageHelper.processMT57_OF_SOS_ALL_NOTIFICATION$REQ_B$TO$Message(userId, json1, null);
                    new Thread(() -> {
                        String a = "sos-" + sosRequest.getT() + "-" + sosRequest.getF() + "-" + messageTime;
                        if (!RedisUtil.exists(a)) {
                            RedisUtil.set(a, String.valueOf(messageTime), 5);
                        }
                        notifyCloseOutTimeSosChannel(a, urlData);
                    }).start();
                    return;
                } finally {
                    sosChannelIds.remove(channelId);
                }

            }
        }
        //抢占失败
        sendSosFail(userId, sosRequest);
    }

    private static void sendSosFail(String userId, CMDSosRequest sosRequest) throws Exception {
        PushResDTO dto2 = new PushResDTO();
        dto2.setGroupId(sosRequest.getT());
        dto2.setType(NotifyEnum.FOURTEEN.getType());
        String json2 = new Gson().toJson(dto2);
        MessageHelper.processMT57_OF_SOS_ALL_NOTIFICATION$REQ_B$TO$Message(userId, json2, null);
    }

    public static void processMT72_OF_ONE_KEY_ALL_NOTIFICATION$REQ_B$TO$Message(CMDSosRequest sosRequest) throws Exception {
        String groupId = sosRequest.getT();
        String[] arr = groupId.split("_");
        String userId = sosRequest.getF();

        Vector<Vector<?>> node = LogicProcessor2.db.queryData("select s_ids,level from chat_group_node where node_id =  '" + arr[1] + "' and base_id = '" + arr[0] + "' and g_status = 1 and node_type = 'MAIN_CHANNEL'");
        String nodeSids = node.get(0).get(0).toString();
        String maxHighLevel = node.get(0).get(1).toString();
        Set<String> keys = RedisUtil.keys(RtcRedisKeyConstants.KEYS_ONE_GROUP(arr[0]));

        if (!keys.isEmpty()) {
            List<String> channelIds = keys.stream().map(a -> a.split(":")[5]).collect(Collectors.toList());
            boolean exist = iterateParentChannels(channelIds, arr[1]);

            // 上级同一部门有人在一键到底,抢占失败
            if (!exist) {
                PushResDTO dto2 = new PushResDTO();
                dto2.setType(NotifyEnum.SIXTEEN.getType());
                dto2.setGroupId(groupId);
                String json2 = new Gson().toJson(dto2);
                MessageHelper.processMT72_OF_ONE_ALL_NOTIFICATION$REQ_B$TO$Message(userId, json2, null);
            } else {
                // 有人在一键到底,但是不是同一部门不影响
                oneKeyList(groupId, userId, maxHighLevel, nodeSids, arr[1]);
            }
        } else {
            // 没有竞争直接抢占成功
            oneKeyList(groupId, userId, maxHighLevel, nodeSids, arr[1]);
            RedisUtil.set("onekey-" + sosRequest.getT(), "temp", 3);
        }
    }

    private static Boolean iterateParentChannels(List<String> channels, String currentChannel) throws Exception {
        if (channels.contains(currentChannel)) {
            return false;
        }
        Vector<Vector<?>> parent = LogicProcessor2.db.queryData("select node_pid from chat_group_node where node_type = 'MAIN_CHANNEL' and g_status = 1 and node_id = '" + currentChannel + "'");
        if (!parent.isEmpty() && EmptyUtils.isNotEmpty(parent.get(0).get(0))) {
            String parentId = parent.get(0).get(0).toString();
            return iterateParentChannels(channels, parentId);
        }
        return true;
    }

    private static void oneKeyList(String groupId, String userId, String level, String nodeSids, String channleId) throws Exception {
        String[] arr = groupId.split("_");
        String sql = "select user_uid from chat_group_node where level >= " + level + " and base_id = '" + arr[0] + "' and node_type = 'USER' and s_ids like '%" + nodeSids + "%' and s_ids like '%," + userId + ",%' and g_status = 1";
        Vector<Vector<?>> vectors = LogicProcessor2.db.queryData(sql);
        String name = getUserName(userId);
        long messageTime = System.currentTimeMillis();
        PushResDTO dto = new PushResDTO();
        String urlData = ConstRtcApp.USER + "-ONEKEY-" + groupId + "-" + level + "-" + userId + "-" + messageTime;
        dto.setUrlData(urlData);
        dto.setAppType(ConstRtcApp.APP_PTT);
        dto.setName(name);
        dto.setPushId(userId);
        dto.setGroupId(groupId);
        dto.setChannelId(channleId);
        dto.setType(NotifyEnum.SEVENTEEN.getType());
        dto.setLevel(level);
        String json = new Gson().toJson(dto);
        for (Vector<?> vector : vectors) {
            String str = vector.get(0).toString();
            if (userId.equals(str)) {
                continue;
            }
            MessageHelper.processMT72_OF_ONE_ALL_NOTIFICATION$REQ_B$TO$Message(str, json, null);
        }

        Map<String, AtomicInteger> numberMap = UserProtocalsType.SOS_NUMBER_MAP;
        waitNotify(groupId, vectors.size(), numberMap);

        PushResDTO dto1 = new PushResDTO();
        dto1.setUrlData(urlData);
        dto1.setAppType(ConstRtcApp.APP_PTT);
        dto1.setType(NotifyEnum.FIVETEEN.getType());
        dto1.setGroupId(groupId);
        dto1.setPushId(userId);
        dto1.setName(name);
        dto1.setLevel(level);
        dto1.setChannelId(channleId);
        String json1 = new Gson().toJson(dto1);
        numberMap.remove(groupId);
        MessageHelper.processMT72_OF_ONE_ALL_NOTIFICATION$REQ_B$TO$Message(userId, json1, null);
    }

    public static void processMT63_OF_SOS_ALL_NOTIFICATION_SUCCESS_REMOVE_KEY(CMDSosRequest sosRequest, String urlData) throws Exception {
        String userId = sosRequest.getF();
        String to = sosRequest.getT();
        String[] split = to.split("_");
        String sql = "select user_uid from chat_group_node where base_id ='" + split[0] + "' and node_pid = '" + split[1] + "' and node_type = 'USER'";
        Vector<Vector<?>> vectors = LogicProcessor2.db.queryData(sql);
        if (EmptyUtils.isNotEmpty(vectors)) {
            for (Vector<?> vector : vectors) {
                String str = vector.get(0).toString();
                if (userId.equals(str)) {
                    continue;
                }
                PushResDTO dto = new PushResDTO();
                dto.setType(NotifyEnum.FOUR.getType());
                dto.setGroupId(to);
                dto.setPushId(userId);
                dto.setUrlData(urlData);
                dto.setChannelId(split[1]);
                String json = new Gson().toJson(dto);
                MessageHelper.processMT63_OF_SOS_ALL_NOTIFICATION_SUCCESS$REQ_B$TO$Message(str, json, null);
            }
        }
        String value = RedisUtil.get(RtcRedisKeyConstants.GET_SOS_GROUP_KEY(split[0], split[1]));
        if (!Objects.isNull(value)) {
            String[] strings = value.split("-");
            if (userId.equals(strings[0])) {
                RedisUtil.del((RtcRedisKeyConstants.GET_SOS_GROUP_KEY(split[0], split[1])));
            }
        }
    }

    public static void processMT72_OF_ONEKEY_ALL_NOTIFICATION_SUCCESS_REMOVE_KEY(CMDSosRequest sosRequest, String currentLevel, String urlData) throws Exception {
        String userId = sosRequest.getF();
        String groupId = sosRequest.getT();
        String[] arr = groupId.split("_");
        Vector<Vector<?>> node = LogicProcessor2.db.queryData("select s_ids from chat_group_node where node_id =  '" + arr[0] + "' and base_id = '" + arr[0] + "' and g_status = 1 and node_type = 'MAIN_CHANNEL'");
        String nodeSids = node.get(0).get(0).toString();
        String sql = "select user_uid from chat_group_node where level > " + currentLevel + " and base_id = '" + arr[0] + "' and node_type = 'USER' and s_ids like '%" + nodeSids + "%' and s_ids like '%," + userId + ",%' and g_status = 1";
        Vector<Vector<?>> vectors = LogicProcessor2.db.queryData(sql);
        if (EmptyUtils.isNotEmpty(vectors)) {
            for (Vector<?> vector : vectors) {
                String str = vector.get(0).toString();
                if (userId.equals(str)) {
                    continue;
                }
                PushResDTO dto = new PushResDTO();
                dto.setType(NotifyEnum.FOUR.getType());
                dto.setGroupId(arr[0]);
                dto.setPushId(userId);
                dto.setChannelId(arr[1]);
                dto.setLevel(currentLevel);
                dto.setUrlData(urlData);
                String json = new Gson().toJson(dto);
                MessageHelper.processMT72_OF_ONE_ALL_NOTIFICATION$REQ_B$TO$Message(str, json, null);
            }
        }
    }

    private static void waitNotify(String key, int size, Map<String, AtomicInteger> numberMap) {
        long oldTime = System.currentTimeMillis();
        while (true) {
            AtomicInteger atomicInteger = numberMap.get(key);
            if (!Objects.isNull(atomicInteger)) {
                int i = atomicInteger.get();
                if (i == size - 1) {
                    break;
                }
            }

            long newTime = System.currentTimeMillis();
            if (newTime - oldTime >= UserProtocalsType.NOTIFY_WAIT_TIME) {
                break;
            }
        }
    }

    public static void processMT58_VIDEO_NOTIFICATION$REQ_B$TO$Message(CMDSosRequest sosRequest) throws Exception {
        String userId = sosRequest.getF();
        String groupId = sosRequest.getT();
        String[] arr = groupId.split("_");
        long time = System.currentTimeMillis();
        String sql = "select user_uid from chat_group_node where base_id ='" + arr[0] + "' and node_pid = '" + arr[1] + "' and g_status = 1 and node_type = 'USER'";
        Vector<Vector<?>> vectors = LogicProcessor2.db.queryData(sql);
        if (!Objects.isNull(vectors) && vectors.size() > 1) {
            for (Vector<?> vector : vectors) {
                String str = vector.get(0).toString();
                if (userId.equals(str)) {
                    continue;
                }

                PushResDTO dto = new PushResDTO();
                String urlData = ConstRtcApp.USER + "-" + groupId + "-" + userId + "-" + time;
                dto.setUrlData(urlData);
                dto.setAppType(ConstRtcApp.APP_LIVE);
                dto.setGroupId(groupId);
                dto.setPushId(userId);
                dto.setType(NotifyEnum.FIVE.getType());
                String json = new Gson().toJson(dto);
                MessageHelper.processMT58_VIDEO_NOTIFICATION$REQ_B$TO$Message(str, json, null);
            }

            PushResDTO dto = new PushResDTO();
            String urlData = ConstRtcApp.USER + "-" + groupId + "-" + userId + "-" + time;
            dto.setUrlData(urlData);
            dto.setAppType(ConstRtcApp.APP_LIVE);
            dto.setType(NotifyEnum.SEVEN.getType());
            String json = new Gson().toJson(dto);
            MessageHelper.processMT58_VIDEO_NOTIFICATION$REQ_B$TO$Message(userId, json, null);
        }
    }

    public static void processMT58_VIDEO_NOTIFICATION$REQ_B$TO$Message_2(CMDSosRequest sosRequest) throws Exception {
        String userId = sosRequest.getF();
        String groupId = sosRequest.getT();
        long time = System.currentTimeMillis();
        PushResDTO dto = new PushResDTO();
        String urlData = ConstRtcApp.USER + "-" + groupId + "-" + userId + "-" + time;
        dto.setUrlData(urlData);
        dto.setAppType(ConstRtcApp.APP_LIVE);
        dto.setType(NotifyEnum.SEVEN.getType());
        String json = new Gson().toJson(dto);
        MessageHelper.processMT58_VIDEO_NOTIFICATION$REQ_B$TO$Message(userId, json, null);
    }

    private static String getUserName(String userId) throws Exception {
        String name = "";
        if (!CommonUtils.isStringEmpty(userId)) {
            String nameSql = "select nickname from g_users where del_flag = 0 and user_uid ='" + userId + "'";
            Vector<Vector<?>> names = LogicProcessor2.db.queryData(nameSql);
            if (!Objects.isNull(names) && !names.isEmpty()) {
                name += names.get(0).get(0);
            }
        }
        return name;
    }
    public static void processMT59_VIDEO_SUCCESS_NOTIFICATION$REQ_B$TO$Message2(CMDSosRequest sosRequest) throws Exception {
        String userId = sosRequest.getF();
        String groupId = sosRequest.getT();
        String[] arr = groupId.split("_");
        String value = RedisUtil.get(RtcRedisKeyConstants.DEVICE_LIVE_KEY(groupId, userId));
        String sql = "select c.user_uid from chat_group_node c left join g_users s on s.user_uid = c.user_uid where s.user_type = 0 and s.del_flag is false and c.base_id ='" + arr[0] + "' and c.node_pid = '" + arr[1] + "' and c.g_status = 1 and c.node_type = 'USER'";

        Vector<Vector<?>> vectors = LogicProcessor2.db.queryData(sql);
        if (!Objects.isNull(vectors) && vectors.size() > 1) {
            for (Vector<?> vector : vectors) {
                String str = vector.get(0).toString();
                if (userId.equals(str)) {
                    continue;
                }

                PushResDTO dto = new PushResDTO();
                String urlData = ConstRtcApp.USER + "-" + groupId + "-" + userId + "-" + value;
                dto.setUrlData(urlData);
                dto.setGroupId(groupId);
                dto.setPushId(userId);
                dto.setType(NotifyEnum.SIX.getType());
                String json = new Gson().toJson(dto);
                MessageHelper.processMT59_VIDEO_SUCCESS_NOTIFICATION$REQ_B$TO$Message(str, json, null);
            }
        }

        if (!CommonUtils.isStringEmpty(value)) {
            Vector<Vector<?>> vector = LogicProcessor2.db.queryData("select user_type from g_users where user_uid = '" + userId + "' and del_flag is false and user_status = 1 and account_type = 1");
            if (vector.get(0).get(0).toString().equals("0")) {
                RedisUtil.del(RtcRedisKeyConstants.DEVICE_LIVE_KEY(groupId, userId));
            } else {
                String mes = RedisUtil.get("device:support:config:" + userId);
                if (EmptyUtils.isNotEmpty(mes)) {
                    com.alibaba.fastjson2.JSONObject object = JSON.parseObject(mes);
                    com.alibaba.fastjson2.JSONObject basic = object.getJSONObject("basic");
                    Integer remotePush = basic.getInteger("remotepush");
                    if (RedisUtil.exists("device:current:config:" + userId)) {
                        String config = RedisUtil.get("device:current:config:" + userId);
                        com.alibaba.fastjson2.JSONObject congigObject = JSON.parseObject(config);
                        Integer autopush = congigObject.getInteger("autopush");
                        if (autopush == 0) {
                            RedisUtil.del(RtcRedisKeyConstants.DEVICE_LIVE_KEY(groupId, userId));
                        }
                    } else {
                        RedisUtil.del(RtcRedisKeyConstants.DEVICE_LIVE_KEY(groupId, userId));
                    }
                }
            }
        }
    }
    public static void processMT59_VIDEO_SUCCESS_NOTIFICATION$REQ_B$TO$Message(CMDSosRequest sosRequest) throws Exception {
        Map<String, String> videoMap = UserProtocalsType.GROUP_VIDEO_MAP;
        String userId = sosRequest.getF();
        String groupId = sosRequest.getT();
        String[] arr = groupId.split("_");
        String key = groupId + "-" + userId;
        String value = videoMap.get(key);
        String sql = "select user_uid from chat_group_node where base_id ='" + arr[0] + "' and g_status = 1 and node_pid = '" + arr[1] + "' and node_type = 'USER'";
        Vector<Vector<?>> vectors = LogicProcessor2.db.queryData(sql);
        if (!Objects.isNull(vectors) && vectors.size() > 1) {
            for (Vector<?> vector : vectors) {
                String str = vector.get(0).toString();
                if (userId.equals(str)) {
                    continue;
                }

                PushResDTO dto = new PushResDTO();
                String urlData = ConstRtcApp.USER + "-" + groupId + "-" + userId + "-" + value;
                dto.setUrlData(urlData);
                dto.setGroupId(groupId);
                dto.setPushId(userId);
                dto.setType(NotifyEnum.SIX.getType());
                String json = new Gson().toJson(dto);
                MessageHelper.processMT59_VIDEO_SUCCESS_NOTIFICATION$REQ_B$TO$Message(str, json, null);
            }
        }

        String time = videoMap.get(key);
        if (!CommonUtils.isStringEmpty(time)) {
            videoMap.remove(key);
        }
    }


    public static void processMT64_OF_GPS_STOP_NOTIFICATION$REQ_B$TO$Message(GpsDTO dto) throws Exception {
        Map<String, List<String>> map = UserProtocalsType.SUBSCRIBE_GPS_MAP;
        List<String> strings = map.get(dto.getGroupId());
        if (Objects.isNull(strings)) {
            map.put(dto.getGroupId(), new ArrayList<>());
            return;
        }
        for (Map.Entry<String, List<String>> entry : map.entrySet()) {
            String key = entry.getKey();
            List<String> list = entry.getValue();

            if (key.equals(dto.getGroupId())) {
                String json = new Gson().toJson(dto);

                for (String userId : list) {
                    MessageHelper.processMT64_OF_GPS_STOP_NOTIFICATION$REQ_B$TO$Message(userId, json, null);
                }
            }
        }
    }

    public static void processMT54_OF_GPS_NOTIFICATION$REQ_B$TO$Message(GpsDTO dto) throws Exception {
        UsersStatusManager liverUsersMgr = ChatServerEventListener.getInstance().getLiverUsersMgr();
        List<String> strList = new ArrayList<>();
        List<Object[]> objList = new ArrayList<>();
        insertUserGps(strList, objList, dto.getUserId(), dto.getGpsLat(), dto.getGpsLng(), dto.getGroupId());
        LogicProcessor2.db.update(strList, objList, true);
        if (EmptyUtils.isNotEmpty(dto.getGroupId())) {
            Map<String, List<String>> map = UserProtocalsType.SUBSCRIBE_GPS_MAP;
            List<String> strings = map.get(dto.getGroupId());
            if (Objects.isNull(strings)) {
                map.put(dto.getGroupId(), new ArrayList<>());
                return;
            }
            for (Map.Entry<String, List<String>> entry : map.entrySet()) {
                String key = entry.getKey();
                List<String> list = entry.getValue();

                if (key.equals(dto.getGroupId())) {
                    String json = new Gson().toJson(dto);

                    for (String userId : list) {
                        if (liverUsersMgr.isOnline(userId)) {
                            MessageHelper.processMT54_OF_GPS_NOTIFICATION$REQ_B$TO$Message(userId, json, null);
                        }
                    }
                }
            }
        }
    }

    public static void insertUserGps(List<String> sqls, List<Object[]> values, String userUid, String lat, String lng, String groupId) {
        sqls.add("INSERT INTO t_gps(gps_uid,gps_lat,gps_lng,group_id" +
                ") VALUES(?,?,?,?)");
        values.add(new Object[]{userUid, lat, lng, groupId});
    }

    public static void processMT60_GPS_SUBSCRIBE$REQ_B$TO$Message(GpsSubscribeDTO dto) throws Exception {
        Map<String, List<String>> map = UserProtocalsType.SUBSCRIBE_GPS_MAP;
        String userId1 = dto.getUserId();

        String key = dto.getGroupId();
        List<String> list = map.get(key);
        if (Objects.isNull(list)) {
            List<String> arrayList = new ArrayList<>();
            arrayList.add(userId1);
            map.put(key, arrayList);
        } else {
            if (!list.contains(userId1)) {
                list.add(userId1);
            }
        }

        PushResDTO sosResDTO = new PushResDTO();
        sosResDTO.setType(NotifyEnum.EIGHT.getType());
        String json = new Gson().toJson(sosResDTO);
        MessageHelper.processMT60_GPS_SUBSCRIBE$$REQ_B$TO$Message(userId1, json, null);
    }

    public static void processMT61_GPS_CANCEL_SUBSCRIBE$REQ_B$TO$Message(GpsSubscribeDTO dto) throws Exception {
        Map<String, List<String>> map = UserProtocalsType.SUBSCRIBE_GPS_MAP;
        String userId = dto.getUserId();
        String key = dto.getGroupId();

        List<String> list = map.get(key);
        if (!Objects.isNull(list) && !list.isEmpty()) {
            for (int i1 = 0; i1 < list.size(); i1++) {
                String d = list.get(i1);
                if (d.equals(userId)) {
                    list.remove(i1);
                    break;
                }
            }
        }

        PushResDTO sosResDTO = new PushResDTO();
        sosResDTO.setType(NotifyEnum.NINE.getType());
        String json = new Gson().toJson(sosResDTO);
        MessageHelper.processMT61_GPS_CANCELSUBSCRIBE$$REQ_B$TO$Message(userId, json, null);
    }

    public static void processMT62_QUERY_VIDEO_SUBSCRIBE$REQ_B$TO$Message(CMDSosRequest res) throws Exception {
        Map<String, String> videoMap = UserProtocalsType.GROUP_VIDEO_MAP;
        String userId = res.getF();
        String groupId = res.getT();
        String id = res.getM();
        String key = groupId + "-" + id;
        String keyone = videoMap.keySet().stream().filter(a -> a.contains(groupId) && a.contains("-" + id)).collect(Collectors.toList()).get(0);
        String value = videoMap.get(keyone);

        PushResDTO dto = new PushResDTO();
        if (CommonUtils.isStringEmpty(value)) {
            dto.setType(NotifyEnum.ELEVEN.getType());
            String json = new Gson().toJson(dto);
            MessageHelper.processMT62_QUERY_VIDEO_SUBSCRIBE$$REQ_B$TO$Message(userId, json, null);
            return;
        }

        String urlData = ConstRtcApp.USER + "-" + keyone + "-" + value;
        logger.error("------------------subvideo:{}----------------", urlData);
        dto.setUrlData(urlData);
        dto.setAppType(ConstRtcApp.APP_LIVE);
        dto.setPushId(id);
        dto.setGroupId(groupId);
        dto.setType(NotifyEnum.TEN.getType());
        String json = new Gson().toJson(dto);
        MessageHelper.processMT62_QUERY_VIDEO_SUBSCRIBE$$REQ_B$TO$Message(userId, json, null);
    }

    public static synchronized void processMT62_QUERY_VIDEO_SUBSCRIBE$REQ_B$TO$Message_2(CMDSosRequest res) throws Exception {
        String userId = res.getF();
        String groupId = res.getT();
        String id = res.getM();
        String deviceType = LogicProcessor2.db.queryData("select user_type from g_users where del_flag = 0 and user_uid = '" + id + "'").get(0).get(0).toString();
        String urlData;
        String node_id = LogicProcessor2.db.queryData("select node_pid from chat_group_node where user_uid = '" + id + "' and node_type = 'USER' and base_id = '" + groupId + "' and g_status = 1").get(0).get(0).toString();
        Set<String> videoMap = RedisUtil.keys(RtcRedisKeyConstants.DEVICE_LIVE_KEYS(groupId));
        if (Objects.equals(deviceType, "1")) {
            long count = videoMap.stream().filter(a -> a.contains(groupId + "_" + node_id) && a.contains(id)).count();
            if (count == 0) {
                logger.info("-------------------------- 00 --------------------------------");

                String mes = RedisUtil.get("device:support:config:" + id);
                if (EmptyUtils.isNotEmpty(mes)) {
                    logger.info("-------------------------- 11 --------------------------------");
                    com.alibaba.fastjson2.JSONObject object = JSON.parseObject(mes);
                    com.alibaba.fastjson2.JSONObject basic = object.getJSONObject("basic");
                    if (basic.containsKey("remotepush") && (basic.getInteger("remotepush") == null || basic.getInteger("remotepush") == 0)) {
                        logger.info("-------------------------- 22 --------------------------------");
                        PushResDTO dto = new PushResDTO();
                        dto.setAppType(ConstRtcApp.APP_LIVE);
                        dto.setType(NotifyEnum.ELEVEN.getType());
                        dto.setPushId(id);
                        MessageHelper.ACTION_NOT_STAND_BY_REMOTE_PUSH(userId, new Gson().toJson(dto), null);
                        return;
                    }
                }
                long value = System.currentTimeMillis();
                PushResDTO dto = new PushResDTO();
                urlData = ConstRtcApp.USER + "-" + groupId + "_" + node_id + "-" + id + "-" + value;
                dto.setUrlData(urlData);
                dto.setAppType(ConstRtcApp.APP_LIVE);
                dto.setType(NotifyEnum.SEVEN.getType());
                dto.setPushId(id);
                MessageHelper.process87ACTION_DEVICE_START_PUSH_STREAM(id, new Gson().toJson(dto), null);
                logger.info("-------------------------- 33 --------------------------------");

                final long time = System.currentTimeMillis();
                while (true) {
                    Set<String> set = RedisUtil.keys(RtcRedisKeyConstants.DEVICE_LIVE_KEYS(groupId));
                    long exist = set.stream().filter(a -> a.contains(groupId + "_" + node_id) && a.contains(id)).count();
                    if (exist > 0) {
                        logger.info("-------------------------- wait device push stream success --------------------------------");
                        break;
                    } else {
                        if (System.currentTimeMillis() - time >= 1000 * 22) {
                            logger.info("-------------------------- 44 --------------------------------");
                            PushResDTO resDTO = new PushResDTO();
                            resDTO.setType(NotifyEnum.ELEVEN.getType());
                            resDTO.setPushId(id);
                            MessageHelper.processMT62_QUERY_VIDEO_SUBSCRIBE$$REQ_B$TO$Message(userId, new Gson().toJson(resDTO), null);
                            return;
                        }
                    }
                    Thread.sleep(200);
                }
            } else {
                logger.info("-------------------------- 55 --------------------------------");
                String value = RedisUtil.get(RtcRedisKeyConstants.DEVICE_LIVE_KEY(groupId + "_" + node_id, id));
                urlData = ConstRtcApp.USER + "-" + groupId + "_" + node_id + "-" + id + "-" + value;
//                PushResDTO dto = new PushResDTO();
//                dto.setUrlData(urlData);
//                dto.setAppType(ConstRtcApp.APP_LIVE);
//                dto.setType(NotifyEnum.SEVEN.getType());
//                MessageHelper.process87ACTION_DEVICE_START_PUSH_STREAM(id, new Gson().toJson(dto), null);
            }
        } else {
            if (Objects.equals(deviceType, "0")) {
                String key = videoMap.stream().filter(a -> a.contains(groupId) && a.contains(id)).findFirst().get();
                String value = RedisUtil.get(key);
                String[] split = key.split(":");
                urlData = ConstRtcApp.USER + "-" + split[3] + "-" + id + "-" + value;
            } else {
                String key = videoMap.stream().filter(a -> a.contains(groupId) && a.contains(id)).findFirst().get();
                urlData = RedisUtil.get(key);
            }
        }
        PushResDTO dto = new PushResDTO();
        logger.error("------------------subvideo:{}----------------", urlData);
        dto.setUrlData(urlData);
        dto.setAppType(ConstRtcApp.APP_LIVE);
        dto.setPushId(id);
        dto.setGroupId(groupId);
        dto.setType(NotifyEnum.TEN.getType());
        String json = new Gson().toJson(dto);
        MessageHelper.processMT62_QUERY_VIDEO_SUBSCRIBE$$REQ_B$TO$Message(userId, json, null);
        // record people who are watching dev live video
        if (Objects.equals(deviceType, "1")) {
            if (RedisUtil.exists(RtcRedisKeyConstants.DEV_LIVE_STREAM_CHANNEL_MEMBER_KEY + id)) {
                String s = RedisUtil.get(RtcRedisKeyConstants.DEV_LIVE_STREAM_CHANNEL_MEMBER_KEY + id);
                JSONArray objects = JSONObject.parseArray(s);
                if (!objects.contains(userId)) {
                    objects.add(userId);
                }
                RedisUtil.set(RtcRedisKeyConstants.DEV_LIVE_STREAM_CHANNEL_MEMBER_KEY + id, objects.toJSONString(), 7200);
            } else {
                JSONArray objects = new JSONArray();
                objects.add(userId);
                RedisUtil.set(RtcRedisKeyConstants.DEV_LIVE_STREAM_CHANNEL_MEMBER_KEY + id, objects.toJSONString(), 7200);
            }
        }
    }

    /**
     * 设备推流通知
     */
    public static void processMT65_QUERY_VIDEO_SUBSCRIBE$REQ_B$TO$Message(CMDSosRequest res) throws Exception {
        String devId = res.getF();
        long time = System.currentTimeMillis();
        String sql = "select structure_level from structure_info WHERE structure_no=(select structure_no from my_structure WHERE dev_id='" + devId + "' and is_finish=0)";
        Vector<Vector<?>> vector = ChatServerUtils.queryUserByDevId(devId);
        if (!vector.isEmpty()) {
            for (Vector<?> v : vector) {
                String str = v.get(0).toString();
                PushResDTO dto = new PushResDTO();
                String urlData = ConstRtcApp.DEV + "-" + devId + "-" + time;
                dto.setUrlData(urlData);
                dto.setPushId(devId);
                dto.setAppType(ConstRtcApp.APP_LIVE);
                dto.setType(NotifyEnum.FIVE.getType());
                String json = new Gson().toJson(dto);
                MessageHelper.processMT65_VIDEO_NOTIFICATION$REQ_B$TO$Message(str, json, null);
            }
        }
        PushResDTO dto = new PushResDTO();
        String urlData = ConstRtcApp.DEV + "-" + devId + "-" + time;
        dto.setUrlData(urlData);
        dto.setAppType(ConstRtcApp.APP_LIVE);
        dto.setType(NotifyEnum.SEVEN.getType());
        String json = new Gson().toJson(dto);
        MessageHelper.processMT65_VIDEO_NOTIFICATION$REQ_B$TO$Message(devId, json, null);
    }

    /**
     * 设备停止推流通知
     */
    public static void processMT66_VIDEO_SUCCESS_NOTIFICATION$REQ_B$TO$Message(CMDSosRequest sosRequest) throws Exception {
        Map<String, String> videoMap = UserProtocalsType.DEV_VIDEO_MAP;
        String devId = sosRequest.getF();
        String value = videoMap.get(devId);
        Vector<Vector<?>> vector = ChatServerUtils.queryUserByDevId(devId);
        if (!vector.isEmpty()) {
            for (Vector<?> v : vector) {
                String str = v.get(0).toString();
                PushResDTO dto = new PushResDTO();
                String urlData = ConstRtcApp.DEV + "-" + devId + "-" + value;
                dto.setUrlData(urlData);
                dto.setPushId(devId);
                dto.setType(NotifyEnum.SIX.getType());
                String json = new Gson().toJson(dto);
                MessageHelper.processMT66_VIDEO_NOTIFICATION$REQ_B$TO$Message(str, json, null);
            }
        }
        String time = videoMap.get(devId);
        if (!CommonUtils.isStringEmpty(time)) {
            videoMap.remove(devId);
        }
    }

    /**
     * 设备上传配置信息通知
     */
    public static void processMT67_DEV_CONFIG_NOTIFICATION$REQ_B$TO$Message(DevConfigDTO configDTO) throws Exception {
        Jedis jedis = RedisManager.getJedis();
        String support = RedisConfig.getSupportKey(configDTO.getDevId());
        String config = RedisConfig.getConfigKey(configDTO.getDevId());
        if (jedis.exists(support)) {
            jedis.getSet(support, JSON.toJSONString(configDTO.getSupport()));
        } else {
            jedis.set(support, JSON.toJSONString(configDTO.getSupport()));
        }
        if (jedis.exists(config)) {
            jedis.getSet(config, JSON.toJSONString(configDTO.getConfig()));
        } else {
            jedis.set(config, JSON.toJSONString(configDTO.getConfig()));
        }
        Vector<Vector<?>> vector = ChatServerUtils.queryUserByDevId(configDTO.getDevId());
        if (!vector.isEmpty()) {
            for (Vector<?> v : vector) {
                String str = v.get(0).toString();
                MessageHelper.processMT67_DEV_CONFIG_NOTIFICATION$REQ_B$TO$Message(str, JSON.toJSONString(configDTO), null);
            }
        }
    }

    public static void processMT74RtcCallPhone(RtcCallPhone rtcCallPhone, String currentUserId) throws Exception {
        CallResDto dto = new CallResDto();
        String code = rtcCallPhone.getCode();
        dto.setAppType(ConstRtcApp.APP_PHONE);
        if (Objects.equals(code, "4")) {
            if (EmptyUtils.isEmpty(rtcCallPhone.getFromUserId())) {
                rtcCallPhone.setFromUserId(currentUserId);
            }
            Vector<Vector<?>> vectors = LogicProcessor2.db.queryData("select nickname from g_users where del_flag = 0 and user_uid = '" + rtcCallPhone.getFromUserId() + "'");
            dto.setFromUserName(vectors.get(0).get(0).toString());
            Vector<Vector<?>> linstener = LogicProcessor2.db.queryData("select nickname from g_users where del_flag = 0 and user_uid = '" + rtcCallPhone.getToUserId() + "'");
            dto.setToUserName(linstener.get(0).get(0).toString());
            boolean exists = RedisUtil.exists("phone-" + rtcCallPhone.getToUserId());
            if (exists) {
                dto.setCode("0");
                dto.setFromUserId(rtcCallPhone.getFromUserId());
                dto.setToUserId(rtcCallPhone.getToUserId());
                MessageHelper.sendPhoneMessage(rtcCallPhone.getFromUserId(), rtcCallPhone.getToUserId(), new Gson().toJson(dto), false, null, UserProtocalsType.MT74_RTC_PHONE_CALL, null);
            } else {
                long st = System.currentTimeMillis();
                String fromUrl = "PHONE-" + rtcCallPhone.getFromUserId() + "-" + rtcCallPhone.getToUserId() + "-" + st;
                String toUrl = "PHONE-" + rtcCallPhone.getToUserId() + "-" + rtcCallPhone.getFromUserId() + "-" + st;
                dto.setFromUserId(rtcCallPhone.getFromUserId());
                dto.setToUserId(rtcCallPhone.getToUserId());
                dto.setFromUrl(fromUrl);
                dto.setToUrl(toUrl);
                dto.setCode("4");
                dto.setStartTime(System.currentTimeMillis());
                MessageHelper.sendPhoneMessage(rtcCallPhone.getFromUserId(), rtcCallPhone.getToUserId(), new Gson().toJson(dto), false, null, UserProtocalsType.MT74_RTC_PHONE_CALL, null);
                dto.setCode("5");
                dto.setFromUrl(toUrl);
                dto.setToUrl(fromUrl);
                MessageHelper.sendPhoneMessage(rtcCallPhone.getToUserId(), rtcCallPhone.getFromUserId(), new Gson().toJson(dto), false, null, UserProtocalsType.MT74_RTC_PHONE_CALL, null);
                callOutimeMap.put(dto.getFromUserId() + "-" + dto.getToUserId(), dto);
            }
        } else {
            try {
                logger.error("-------------------rtcCallPhone-----------------:{}", JSON.toJSONString(rtcCallPhone));
                dto.setFromUserId(rtcCallPhone.getFromUserId());
                dto.setToUserId(rtcCallPhone.getToUserId());
                dto.setCode(rtcCallPhone.getCode());
                dto.setOperateUserId(currentUserId);
                if (Objects.equals(code, "1") && Objects.equals(currentUserId, rtcCallPhone.getFromUserId())) {
                    MessageHelper.sendPhoneMessage(rtcCallPhone.getToUserId(), rtcCallPhone.getFromUserId(), new Gson().toJson(dto), false, null, UserProtocalsType.MT74_RTC_PHONE_CALL, null);
                } else {
                    MessageHelper.sendPhoneMessage(rtcCallPhone.getFromUserId(), rtcCallPhone.getToUserId(), new Gson().toJson(dto), false, null, UserProtocalsType.MT74_RTC_PHONE_CALL, null);
                }
                if (Objects.equals(dto.getCode(), "2")) {
                    RedisUtil.set("phone-" + rtcCallPhone.getToUserId(), dto.getCode(), 1800);
                    callTimeMap.put(dto.getFromUserId() + "-" + dto.getToUserId(), System.currentTimeMillis());
                }
                if (Objects.equals(dto.getCode(), "1")) {
                    RedisUtil.del("phone-" + rtcCallPhone.getToUserId());
                }
                if (!Objects.equals(dto.getCode(), "2")) {
                    insertCallMsgs(dto);
                }
            } finally {
                callOutimeMap.remove(rtcCallPhone.getFromUserId() + "-" + rtcCallPhone.getToUserId());
            }
        }
    }

    private static void insertCallMsgs(CallResDto dto) {
        try {
            String key = dto.getFromUserId() + "-" + dto.getToUserId();
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("code", dto.getCode());
            jsonObject.put("operateUserId", dto.getOperateUserId());
            jsonObject.put("fromUserId", dto.getFromUserId());
            jsonObject.put("toUserId", dto.getToUserId());
            if (callTimeMap.containsKey(key)) {
                jsonObject.put("duration", System.currentTimeMillis() - callTimeMap.get(key));
                callTimeMap.remove(key);
            }
            final String fp = UUID.randomUUID().toString();
            MsgBody4Group ms = new MsgBody4Group();
            ms.setM(jsonObject.toJSONString());
            ms.setTy(11);
            ms.setCy(0);
            ms.setF(dto.getFromUserId());
            ms.setT(dto.getToUserId());
            new Thread(() -> {
                try {
                    Vector<Vector<?>> vectors = LogicProcessor2.db.queryData("select nickname from g_users where del_flag = 0 and user_uid = '" + dto.getFromUserId() + "'");
                    Vector<Vector<?>> listener = LogicProcessor2.db.queryData("select nickname from g_users where del_flag = 0 and user_uid = '" + dto.getToUserId() + "'");
                    ms.setNickName(vectors.get(0).get(0).toString());
                    MessageHelper.sendPhoneMessage(dto.getToUserId(), dto.getFromUserId(), JSONObject.toJSONString(ms), true, UserProtocalsType.MT03_OF_CHATTING_MESSAGE, null, fp);
                    ms.setNickName(listener.get(0).get(0).toString());
                    MessageHelper.sendPhoneMessage(dto.getFromUserId(), dto.getToUserId(), JSONObject.toJSONString(ms), true, UserProtocalsType.MT03_OF_CHATTING_MESSAGE, null, fp);
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                }
            }).start();
            new Thread(() -> {
                try {
                    LogicProcessor2.db.update("INSERT INTO chat_msgs_collect(src_uid,dest_uid,chat_type,msg_type,msg_content,online_count,msg_time, msg_time2,fingerprint)" +
                                    " VALUES(?,?,?,?,?,?," + DBDepend.getDefaultDatetimeFunc() + "," + System.currentTimeMillis() + ",?)"
                            , new Object[]{dto.getFromUserId(), dto.getToUserId(), "0", "11", jsonObject.toJSONString(), 2, fp}, false);
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                }
            }).start();
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    public static void processACTION_DEVICE_START_PUSH_STREAM(CMDSosRequest sosRequest, String userId) throws Exception {
        String groupId = sosRequest.getT();
        long time = System.currentTimeMillis();
        PushResDTO dto = new PushResDTO();
        String urlData = ConstRtcApp.USER + "-" + groupId + "-" + userId + "-" + time;
        dto.setUrlData(urlData);
        dto.setAppType(ConstRtcApp.APP_LIVE);
        dto.setType(NotifyEnum.SEVEN.getType());
        String json = new Gson().toJson(dto);
        MessageHelper.processMT58_VIDEO_NOTIFICATION$REQ_B$TO$Message(userId, json, null);
    }

    public static void process88ACTION_DEVICE_STOP_PUSH_STREAM(CMDSosRequest sosRequest, String userId) throws Exception {
        String groupId = sosRequest.getT();
        String[] arr = groupId.split("_");
        String value = RedisUtil.get(RtcRedisKeyConstants.DEVICE_LIVE_KEY(groupId, userId));
        String sql = "select c.user_uid from chat_group_node c left join g_users s on s.user_uid = c.user_uid where s.user_type = 0 and s.del_flag is false and c.base_id ='" + arr[0] + "' and c.node_pid = '" + arr[1] + "' and c.g_status = 1 and c.node_type = 'USER'";

        Vector<Vector<?>> vectors = LogicProcessor2.db.queryData(sql);
        if (!Objects.isNull(vectors) && vectors.size() > 1) {
            for (Vector<?> vector : vectors) {
                String str = vector.get(0).toString();
                if (userId.equals(str)) {
                    continue;
                }

                PushResDTO dto = new PushResDTO();
                String urlData = ConstRtcApp.USER + "-" + groupId + "-" + userId + "-" + value;
                dto.setUrlData(urlData);
                dto.setGroupId(groupId);
                dto.setPushId(userId);
                dto.setType(NotifyEnum.SIX.getType());
                String json = new Gson().toJson(dto);
                MessageHelper.processMT59_VIDEO_SUCCESS_NOTIFICATION$REQ_B$TO$Message(str, json, null);
            }
        }

        if (!CommonUtils.isStringEmpty(value)) {
            String mes = RedisUtil.get("device:support:config:" + userId);
            if (EmptyUtils.isNotEmpty(mes)) {
                com.alibaba.fastjson2.JSONObject object = JSON.parseObject(mes);
                com.alibaba.fastjson2.JSONObject basic = object.getJSONObject("basic");
                Integer remotePush = basic.getInteger("remotepush");
                if (RedisUtil.exists("device:current:config:" + userId)) {
                    String config = RedisUtil.get("device:current:config:" + userId);
                    com.alibaba.fastjson2.JSONObject congigObject = JSON.parseObject(config);
                    Integer autopush = congigObject.getInteger("autopush");
                    if (autopush == 0) {
                        RedisUtil.del(RtcRedisKeyConstants.DEVICE_LIVE_KEY(groupId, userId));
                    }
                } else {
                    RedisUtil.del(RtcRedisKeyConstants.DEVICE_LIVE_KEY(groupId, userId));
                }
            }
        }
    }

    public static void processMT82_DEVICE_UPLOAD_NETWORK_ENV(String data, String userUid) {
        RedisUtil.set(RtcRedisKeyConstants.DEV_ENV_KEY + userUid, data);
        logger.info("processMT82_DEVICE_UPLOAD_NETWORK_ENV:{}", data);
    }

    // notify dev close video stream
    public static void processACTION_NOTIFY_DEVICE_END_PUSH_STREAM(String userUid, String closeUid) throws Exception {
        Vector<Vector<?>> res = LogicProcessor2.db.queryData("select base_id,node_pid from chat_group_node  where user_uid = '" + userUid + "' and node_type = 'USER' and g_status = 1");
        if (!res.isEmpty()) {
            if (RedisUtil.exists(RtcRedisKeyConstants.DEV_LIVE_STREAM_CHANNEL_MEMBER_KEY + userUid)) {
                String s = RedisUtil.get(RtcRedisKeyConstants.DEV_LIVE_STREAM_CHANNEL_MEMBER_KEY + userUid);
                JSONArray members = JSONObject.parseArray(s);
                if (!members.isEmpty()) {
                    members.remove(closeUid);
                    if (members.isEmpty()) {
                        notifyDevCloseStream(res, userUid);
                    }
                    RedisUtil.set(RtcRedisKeyConstants.DEV_LIVE_STREAM_CHANNEL_MEMBER_KEY + userUid, members.toJSONString(), 7200);
                } else {
                    notifyDevCloseStream(res, userUid);
                }
            } else {
                notifyDevCloseStream(res, userUid);
            }
        }
    }

    private static void notifyDevCloseStream(Vector<Vector<?>> res, String userUid) throws Exception {
        String groupId = res.get(0).get(0).toString();
        String nId = res.get(0).get(1).toString();
        String key = groupId + "_" + nId + "-" + userUid;
        String value = RedisUtil.get(RtcRedisKeyConstants.DEVICE_LIVE_KEY(groupId + "_" + nId, userUid));
        PushResDTO dto = new PushResDTO();
        String urlData = ConstRtcApp.USER + "-" + key + "-" + value;
        dto.setUrlData(urlData);
        dto.setGroupId(groupId);
        dto.setPushId(userUid);
        dto.setType(NotifyEnum.SIX.getType());
        MessageHelper.processACTION_DEVICE_STOP_PUSH_STREAM(userUid, new Gson().toJson(dto), null);
        if (!CommonUtils.isStringEmpty(value)) {
            String mes = RedisUtil.get("device:support:config:" + userUid);
            if (EmptyUtils.isNotEmpty(mes)) {
                com.alibaba.fastjson2.JSONObject object = JSON.parseObject(mes);
                com.alibaba.fastjson2.JSONObject basic = object.getJSONObject("basic");
                Integer remotePush = basic.getInteger("remotepush");
                if (RedisUtil.exists("device:current:config:" + userUid)) {
                    String config = RedisUtil.get("device:current:config:" + userUid);
                    com.alibaba.fastjson2.JSONObject congigObject = JSON.parseObject(config);
                    Integer autopush = congigObject.getInteger("autopush");
                    if ((remotePush == null || remotePush == 1) && autopush == 0) {
                        RedisUtil.del(RtcRedisKeyConstants.DEVICE_LIVE_KEY(groupId + "_" + nId, userUid));
                    }
                } else {
                    RedisUtil.del(RtcRedisKeyConstants.DEVICE_LIVE_KEY(groupId + "_" + nId, userUid));
                }
            }
        }
    }

    //
    public static void recordWhoWatchingDevLive(CMDSosRequest id, String userId) throws Exception {
        // recording people who are watching dev live video
        if (RedisUtil.exists(RtcRedisKeyConstants.DEV_LIVE_STREAM_CHANNEL_MEMBER_KEY + id.getM())) {
            String s = RedisUtil.get(RtcRedisKeyConstants.DEV_LIVE_STREAM_CHANNEL_MEMBER_KEY + id.getM());
            JSONArray objects = JSONObject.parseArray(s);
            if (!objects.contains(userId)) {
                objects.add(userId);
            }
            RedisUtil.set(RtcRedisKeyConstants.DEV_LIVE_STREAM_CHANNEL_MEMBER_KEY + id.getM(), objects.toJSONString(), 7200);
        } else {
            JSONArray objects = new JSONArray();
            objects.add(userId);
            RedisUtil.set(RtcRedisKeyConstants.DEV_LIVE_STREAM_CHANNEL_MEMBER_KEY + id.getM(), objects.toJSONString(), 7200);
        }
    }

    public static void uploadLiveSource(String message, String devId) {
        RedisUtil.set("device:live:source:" + devId, message);
    }

    public static void queryLiveSource(String devId, String fromUseId) throws Exception {
        String message = RedisUtil.get("device:live:source:" + devId);
        MessageHelper.sendMessage(fromUseId, message, false, UserProtocalsType.ACTION_CHANGE_ANDROID_LIVE_SOURCE, null);
    }

    public static void changeLiveSource(String message, String devId) throws Exception {
        MessageHelper.sendMessage(devId, message, false, UserProtocalsType.ACTION_CHANGE_ANDROID_LIVE_SOURCE, null);
    }

    public static void deviceInvestigation(String fromUserId) throws Exception {
        Vector<Vector<?>> va = LogicProcessor2.db.queryData("select n.node_pid,g.nickname,n.base_id from chat_group_node n left join g_users g on g.user_uid = n.user_uid  where g.del_flag is false and g.user_type = 1 and n.user_uid = '" + fromUserId + "' and n.node_type = 'USER' and n.g_status = 1");
        if (!va.isEmpty()) {
            String nodeId = va.get(0).get(0).toString();
            String nickname = va.get(0).get(1).toString();
            String baseId = va.get(0).get(2).toString();
            Vector<Vector<?>> vector = LogicProcessor2.db.queryData("select c.user_uid from chat_group_node c " +
                    " left join g_users g on g.user_uid = c.user_uid " +
                    " where g.user_status = 1 and g.user_type = 0 and g.del_flag is false and c.g_status = 1 and c.node_type = 'USER' and c.user_uid != '" + fromUserId + "' and c.node_pid = '" + nodeId + "'");
            if (!vector.isEmpty()) {
                String message = "设备【" + nickname + "】监测到有移动目标";
                MsgBody4Group msgBody4Group = new MsgBody4Group();
                msgBody4Group.setTy(MsgType.TYPE_MOVE);
                msgBody4Group.setParentFp(Protocal.genFingerPrint());
                msgBody4Group.setT(baseId + "_" + nodeId);
                msgBody4Group.setF(fromUserId);
                msgBody4Group.setM(message);
                msgBody4Group.setNickName(LogicProcessor2.db.queryData("select nickname from g_users where user_uid = '" + fromUserId + "' and del_flag is false").get(0).get(0).toString());
                String time = String.valueOf(System.currentTimeMillis());
                msgBody4Group.setM3(time);
                for (Vector<?> v : vector) {
                    String userId = v.get(0).toString();
                    boolean isOnline = ChatServerEventListener.getInstance().getLiverUsersMgr().isOnline(userId);
                    MessageHelper.SendResultObserver sendResultObserver = (code, p) -> {
                        if (code) {
                            if (!isOnline && ChatServerLauncher.bridgeEnabled) {
                                OfflineMessageManager.processOfflineMessage(p, false, time);
                            }
                        } else {
                            OfflineMessageManager.processOfflineMessage(p, false, time);
                        }
                    };
                    MessageHelper.sendGroupPttMsg_SERVER$TO$B_Message(userId, msgBody4Group, sendResultObserver);
                }
            }
        }
    }

    public static void bindHumanFace(String devId, String id, String groupId) throws Exception {
        Vector<Vector<?>> res = LogicProcessor2.db.queryData("select file_name from human_face_file where id = '" + id + "'");
        if (!res.isEmpty()) {
            MessageHelper.sendMessage(devId, res.get(0).get(0).toString(), false, UserProtocalsType.ACTION_DEVICE_HUMAN_FACE_BIND, null);
        }
    }

    public static void indentifyHumanFace(String cmdSosRequest, String fromUserId) throws Exception {
        JSONObject file = JSONObject.parseObject(cmdSosRequest);
        String faceFileName = file.getString("faceFileName");
        String realFileName = file.getString("realFileName");
        Integer fileType = file.getInteger("fileType");
        String print = Protocal.genFingerPrint();
        Vector<Vector<?>> va = LogicProcessor2.db.queryData("select n.node_pid,n.base_id,g.nickname from chat_group_node n left join g_users g on g.user_uid = n.user_uid  where g.del_flag is false and g.user_type = 1 and n.user_uid = '" + fromUserId + "' and n.node_type = 'USER' and n.g_status = 1");
        if (!va.isEmpty()) {
            String nodeId = va.get(0).get(0).toString();
            String baseId = va.get(0).get(1).toString();
            Vector<Vector<?>> vector = LogicProcessor2.db.queryData("select c.user_uid from chat_group_node c " +
                    " left join g_users g on g.user_uid = c.user_uid " +
                    "where g.user_status = 1 and g.user_type = 0 and g.del_flag is false and c.g_status = 1 and c.node_type = 'USER' and c.user_uid != '" + fromUserId + "' and c.node_pid = '" + nodeId + "'");
            Vector<Vector<?>> result = LogicProcessor2.db.queryData("select username,id_card,sex,number from human_face_file where " + (fileType == 1 ? "file_name" : "number") + " like '%" + faceFileName + "%'");

            JSONObject json = new JSONObject();
            json.put("username", result.get(0).get(0).toString());
            json.put("idCard", result.get(0).get(1).toString());
            json.put("sex", result.get(0).get(2).equals("1"));
            if (fileType == 2) {
                json.put("number", result.get(0).get(3).toString());
            }

            json.put("time", System.currentTimeMillis());
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("message", json);
            jsonObject.put("faceFileName", faceFileName);
            jsonObject.put("realFileName", realFileName);
            if (!vector.isEmpty()) {
                MsgBody4Group msgBody4Group = new MsgBody4Group();
                msgBody4Group.setTy(fileType == 2 ? MsgType.Type_NUMBER : MsgType.TYPE_HUMAN_FACE);
                msgBody4Group.setParentFp(print);
                msgBody4Group.setT(baseId + "_" + nodeId);
                msgBody4Group.setF(fromUserId);
                msgBody4Group.setM(jsonObject.toJSONString());
                msgBody4Group.setNickName(LogicProcessor2.db.queryData("select nickname from g_users where user_uid = '" + fromUserId + "' and del_flag is false").get(0).get(0).toString());
                String time = String.valueOf(System.currentTimeMillis());
                msgBody4Group.setM3(time);
                for (Vector<?> v : vector) {
                    String userId = v.get(0).toString();
                    boolean isOnline = ChatServerEventListener.getInstance().getLiverUsersMgr().isOnline(userId);
                    MessageHelper.SendResultObserver sendResultObserver = (code, p) -> {
                        if (code) {
                            if (!isOnline && ChatServerLauncher.bridgeEnabled) {
                                OfflineMessageManager.processOfflineMessage(p, false, time);
                            }
                        } else {
                            OfflineMessageManager.processOfflineMessage(p, false, time);
                        }
                    };
                    MessageHelper.sendGroupPttMsg_SERVER$TO$B_Message(userId, msgBody4Group, sendResultObserver);
                }
            }
            MessageRecordManager.colectMsgAsync(fromUserId, baseId + "_" + nodeId, MsgBodyRoot.CHAT_TYPE_GROUP$CHAT, String.valueOf(MsgBodyRoot.TYPE_HUMAN_FACE), jsonObject.toString(), print, print);
        }
    }

    public static void deviceUploadFileList(String dataContent, String userId) throws Exception {
        logger.info("-------------------deviceUploadFileList:{}-------------------", dataContent);
        if (EmptyUtils.isNotEmpty(dataContent)) {
            // RedisUtil.set(RtcRedisKeyConstants.DEV_FILE_LIST(userId),dataContent);
            JSONObject jsonObject = JSONObject.parseObject(dataContent);
            if (jsonObject.containsKey("pushId")) {
                String pushId = jsonObject.getString("pushId");
                jsonObject.put("devId", userId);
                MessageHelper.processACTION_PUSH_DEVICE_FILE_PAGE_DATA(pushId, jsonObject.toJSONString(), null);
            }
        }
    }

    public static void uploadFileProgress(String dataContent, String userId) throws Exception {
        logger.info("-----------------progress:{}----------------", dataContent);
        Vector<Vector<?>> vector = LogicProcessor2.db.queryData("select n.user_uid from chat_group_node n left join g_users g on g.user_uid = n.user_uid where n.node_type = 'USER' and g.del_flag is false and g.user_type = 0 and n.base_id = (select base_id from chat_group_node where user_uid = '" + userId + "' and node_type = 'USER' and g_status = 1)");
        if (!vector.isEmpty()) {
            for (Vector<?> v : vector) {
                String userUid = v.get(0).toString();
                MessageHelper.processACTION_DEVICE_FILE_PROGRESS(userUid, dataContent, null);
            }
        }
    }

}
